package com.ruoyi.flink.dataStream;


import com.ruoyi.flink.bean.WindowTxt;
import com.ruoyi.flink.utils.ExerciseBase;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple4;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.AssignerWithPeriodicWatermarks;
import org.apache.flink.streaming.api.functions.windowing.ProcessAllWindowFunction;
import org.apache.flink.streaming.api.watermark.Watermark;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import javax.annotation.Nullable;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;

/**
 * @program: ruoyi
 * @description: 窗口功能测试
 * @author: zengchen
 * @create: 2020-11-28 15:45
 **/
public class WindowTestBak {
    public static void main(String[] args) throws Exception {
        DateTimeFormatter dateTimeFormatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        System.out.println("ExerciseBase.parallelism=" + ExerciseBase.parallelism);
//        env.setParallelism(ExerciseBase.parallelism);
        env.setParallelism(1);
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime); // 设置使用时间，默认是处理时间

//        DataStreamSource<WindowTxt> windowTxtDataStreamSource = env.addSource(new MyTxtSource());

        DataStreamSource<String> socketString = env.socketTextStream("localhost", 9999);
        SingleOutputStreamOperator<WindowTxt> windowTxtDataStreamSource = socketString.map((MapFunction<String, WindowTxt>) value -> {
            WindowTxt windowTxt = new WindowTxt();
            String[] s = value.split(" ");
            windowTxt.setKey(s[0]);
            windowTxt.setDate(s[1]);
            windowTxt.setTime(s[2]);
            windowTxt.setValue(Integer.parseInt(s[3]));
            windowTxt.setEventTime(LocalDateTime.parse(s[1] + " " + s[2], dateTimeFormatter));
            return windowTxt;
        });


        SingleOutputStreamOperator<Tuple4<Long, Long, Long, String>> windows = windowTxtDataStreamSource
                .keyBy("key")
                .assignTimestampsAndWatermarks(new TimeWatermark())
                .timeWindowAll(Time.seconds(10))
//                .window(TumblingEventTimeWindows.of(Time.seconds(3)))
                .process(new ProcessAllWindowFunction<WindowTxt, Tuple4<Long, Long, Long, String>, TimeWindow>() {
                    @Override
                    public void process(Context context, Iterable<WindowTxt> elements, Collector<Tuple4<Long, Long, Long, String>> out) throws Exception {
//                        StringBuffer elementsStr = new StringBuffer();
//                        for (WindowTxt ele : elements) {
//                            elementsStr.append(ele).append("/n");
//                        }
//                        out.collect(Tuple4.of(context.window().getStart(), context.window().getEnd(), context.currentWatermark(), elementsStr.toString()));
//                        System.out.println("zmhs:" +context.window().getStart());
//                        System.out.println(context.window().getEnd());
//                        System.out.println(context.currentWatermark());
                    }
                });
        System.out.println("result print ------- ");
        windows.print();
        env.execute("txt test (java)");
    }

    static class TimeWatermark implements AssignerWithPeriodicWatermarks<WindowTxt> {

        @Nullable
        @Override
        public Watermark getCurrentWatermark() {
            return new Watermark(System.currentTimeMillis());
        }


        @Override
        public long extractTimestamp(WindowTxt element, long previousElementTimestamp) {
            return element.getEventTimeStamp();
        }
    }
}
